package script

import (
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"sync"
	"time"
)

// Message 表示消息结构
type Message struct {
	Body    []byte
	Headers map[string]interface{}
}

// Config 连接配置
type Config struct {
	Host     string
	Port     int
	Username string
	Password string
	VHost    string
}

// MQ 客户端接口
type MQ interface {
	NewProducer() (Producer, error)
	NewConsumer(queue string) (Consumer, error)
	Close() error
}

// Producer 生产者接口
type Producer interface {
	Publish(msg Message) error
}

// Consumer 消费者接口
type Consumer interface {
	Start(handler func(Message) bool) error
	Stop() error
}

// 错误定义
var (
	ErrNotConnected    = errors.New("not connected to MQ server")
	ErrInvalidConfig   = errors.New("invalid MQ configuration")
	ErrConsumerRunning = errors.New("consumer is already running")
	ErrConsumerStopped = errors.New("consumer is not running")
)

// mqClient 实现MQ接口
type mqClient struct {
	config    Config
	queue     string //队列名称
	connected bool   //标记链接
	mu        sync.Mutex
	// 实际的MQ连接对象可以在这里定义
	conn *amqp.Connection //与MQ服务器的底层TCP连接对象
	ch   *amqp.Channel    //是通过连接创建的通道对象
}

// mqProducer 实现Producer接口
type mqProducer struct {
	client *mqClient
}

// mqConsumer 实现Consumer接口
type mqConsumer struct {
	client   *mqClient
	queue    string
	running  bool //执行
	mu       sync.Mutex
	quitChan chan struct{}
	wg       sync.WaitGroup
}

// New 创建新的MQ客户端
func New(config Config, queue string) (MQ, error) {
	if config.Host == "" || config.Port <= 0 || queue == "" {
		return nil, ErrInvalidConfig
	}

	client := &mqClient{
		config: config,
		queue:  queue,
	}

	if err := client.connect(); err != nil {
		return nil, fmt.Errorf("connection failed: %w", err)
	}

	return client, nil
}

// 连接到MQ服务器
func (c *mqClient) connect() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	//实际连接代码示例(RabbitMQ)：
	addr := fmt.Sprintf("amqp://%s:%s@%s:%d/%s",
		c.config.Username, c.config.Password,
		c.config.Host, c.config.Port, c.config.VHost)
	conn, err := amqp.Dial(addr)
	if err != nil {
		return err
	}
	c.conn = conn
	c.ch, err = conn.Channel()
	if err != nil {
		return err
	}

	// 声明队列
	_, err = c.ch.QueueDeclare(
		c.queue, // name
		true,    // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)

	c.connected = true
	fmt.Printf("Connected to MQ server: %s:%d, queue: %s\n",
		c.config.Host, c.config.Port, c.queue)
	return nil
}

// NewProducer 创建生产者
func (c *mqClient) NewProducer() (Producer, error) {
	if !c.connected {
		return nil, ErrNotConnected
	}
	return &mqProducer{client: c}, nil
}

// NewConsumer 创建消费者
func (c *mqClient) NewConsumer(queue string) (Consumer, error) {
	if !c.connected {
		return nil, ErrNotConnected
	}
	return &mqConsumer{
		client:   c,
		queue:    queue,
		quitChan: make(chan struct{}),
	}, nil
}

// Close 关闭连接
func (c *mqClient) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.connected {
		return nil
	}

	// 实际关闭连接代码
	if err := c.ch.Close(); err != nil {
		return err
	}
	if err := c.conn.Close(); err != nil {
		return err
	}

	c.connected = false
	fmt.Println("MQ connection closed")
	return nil
}

// Publish 发送消息
func (p *mqProducer) Publish(msg Message) error {
	p.client.mu.Lock()
	defer p.client.mu.Unlock()

	if !p.client.connected {
		return ErrNotConnected
	}

	//实际发布消息代码
	err := p.client.ch.Publish(
		"",             // exchange
		p.client.queue, // routing key
		false,          // mandatory
		false,          // immediate
		amqp.Publishing{
			ContentType: "application/octet-stream",
			Body:        msg.Body,
			Headers:     msg.Headers,
		},
	)
	if err != nil {
		return err
	}

	fmt.Printf("Published message to %s: %s\n", p.client.queue, string(msg.Body))
	return nil
}

// Start 开始消费消息
func (c *mqConsumer) Start(handler func(Message) bool) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.running {
		return ErrConsumerRunning
	}

	c.running = true
	c.wg.Add(1)

	go func() {
		defer c.wg.Done()
		c.consumeLoop(handler)
	}()

	fmt.Printf("Started consuming from %s\n", c.queue)
	return nil
}

// 消费循环，处理消息并监听退出信号
func (c *mqConsumer) consumeLoop(handler func(Message) bool) {
	// 实际消费代码示例
	msgs, err := c.client.ch.Consume(
		c.queue, // queue
		"",      // consumer
		false,   // auto-ack：关闭自动确认
		false,   // exclusive
		false,   // no-local
		false,   // no-wait
		nil,     // args
	)
	if err != nil {
		fmt.Printf("Consume error: %v\n", err)
		return
	}

	for {
		select {
		case <-c.quitChan:
			fmt.Printf("Stopping consumer for %s\n", c.queue)
			return
		case amqpMsg := <-msgs: // 变量名改为 amqpMsg，避免和自定义 Message 冲突
			// . 调用自定义 handler 处理消息
			continueConsume := handler(Message{
				Body:    amqpMsg.Body,
				Headers: amqpMsg.Headers,
			})
			// 2. 手动确认消息（关键！）
			if err := amqpMsg.Ack(false); err != nil {
				fmt.Printf("Ack message failed: %v\n", err)
			}
			// 3. 若 handler 返回 false，停止消费
			if !continueConsume {
				fmt.Printf("Handler requested stop for %s\n", c.queue)
				return
			}
		}
	}
}

// Stop 停止消费
func (c *mqConsumer) Stop() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.running {
		return ErrConsumerStopped
	}

	close(c.quitChan)
	c.wg.Wait()
	c.running = false

	fmt.Printf("Stopped consumer for %s\n", c.queue)
	return nil
}

// 模拟消息生产(实际使用中移除)
func (c *mqConsumer) simulateMessages(msgs chan<- Message) {
	defer close(msgs)

	count := 0
	for {
		select {
		case <-c.quitChan:
			return
		default:
			msgs <- Message{
				Body: []byte(fmt.Sprintf("Simulated message %d", count)),
				Headers: map[string]interface{}{
					"timestamp": time.Now().Unix(),
				},
			}
			count++
			time.Sleep(2 * time.Second)
		}
	}
}
